package org.example.data_stream;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.List;

/**
 * @author shenguangyang
 */
public class E06_TransformFlatMap {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        List<Event> events = new ArrayList<Event>() {{
            add(new Event("user02", "./cart", 1660095536L));
            add(new Event("user01", "./home", 1660095536L));
        }};
        DataStream<Event> streamSource = env.fromCollection(events);

        streamSource.flatMap(new MyFlatMapper()).print("1");

        streamSource.flatMap((Event value, Collector<String> out)-> {
            out.collect(value.getUser());
            if (value.getUser().equals("user01")) {
                out.collect(value.getUrl());
            }
        }).returns(new TypeHint<String>() {}).print("2");

        env.execute();
    }

    /**
     * FlatMapFunction泛型
     * T: 输入泛型
     * O: 输出泛型
     *
     * Tuple2: flink实现的二元组
     */
    public static class MyFlatMapper implements FlatMapFunction<Event, String> {
        @Override
        public void flatMap(Event value, Collector<String> out) throws Exception {
            out.collect(value.getUser());
            out.collect(value.getUrl());
        }
    }
}
